package com.lvmama.rhino.message;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.apache.log4j.Logger;

import java.util.Properties;

/**
 * Created by wuyunpeng on 2016/4/26.
 */
public class RhinoMessageProducer {

    private static final Logger LOGGER = Logger.getLogger(RhinoMessageProducer.class);

    private static Producer<String, RhinoMessage> producer ;

    private static void init (Properties props){
        ProducerConfig config = new ProducerConfig(props);
        producer = new Producer<String, RhinoMessage>(config);

    }

    public RhinoMessageProducer(Properties props){
        init(props);
    }

    /**
     *
     * @param message 消息
     * @param topic Kafka主题
     */
    public void send(RhinoMessage message,String topic){
        try {
            KeyedMessage<String, RhinoMessage> km = new KeyedMessage<String, RhinoMessage>(topic,message);
            producer.send(km);
        } catch (Exception e) {
            e.printStackTrace();
            LOGGER.info("Send message"+message.toString()+"failed");
        }
    }

    /**
     * 关闭生产者
     */
    public void shutdown() {
        if (producer != null) producer.close();
    }
}
